package com.xinxing.learning.crud.batch.service.impl;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.xinxing.learning.crud.batch.common.ExecutorConfig;
import com.xinxing.learning.crud.batch.common.SqlContext;
import com.xinxing.learning.crud.batch.entity.User;
import com.xinxing.learning.crud.batch.mapper.UserMapper;
import com.xinxing.learning.crud.batch.service.UserService;
import com.xinxing.learning.crud.batch.transactional.CallBack;
import com.xinxing.learning.crud.batch.transactional.TransactionMultipartExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.session.SqlSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

@Slf4j
@Service
public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements UserService {

    @Autowired
    private UserMapper userMapper;

    @Autowired
    private SqlContext sqlContext;

    @Transactional(rollbackFor = Exception.class)
    @Override
    public int batchInsertUser(List<User> userList) {
        int size = userMapper.batchInsertUser(userList);
        return size;
    }

    @Override
    public void multipartThreadBatchInsertUser(List<User> userList) throws InterruptedException {

        TransactionMultipartExecutor<User> userTransactionMultipartExecutor = new TransactionMultipartExecutor<>(3, userList);
        userTransactionMultipartExecutor.setCallback(
                (CallBack<User>) list -> {
                    // for (User user : list) {
                    //     if ("万里晴空9".equals(user.getUsername())){
                    //         user.setUsername("大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9大梦初醒9");
                    //     }
                    // }
                    int size = userMapper.batchInsertUser(list);
                    log.info("插入数据条数:{}", size);
                }
        );
        userTransactionMultipartExecutor.excute();
    }

    /**
     * 通过 SqlSession 控制事务 mybatis 编程
     * 这样事务注解就失效了 测试类就无法控制事务了 可以在测试类的@BeforeEach方法中让测试方法和业务方法共用一个connection
     *
     * @param userList
     * @throws SQLException
     */
    @Transactional
    @Override
    public void multipartThreadBatchInsertUserOnSqlSession(List<User> userList) throws SQLException {
        // 获取数据库连接 获取会话 回到了使用mybatis的变成的方式了
        SqlSession sqlSession = sqlContext.getSqlSession();
        Connection connection = sqlSession.getConnection();
        try {
            // 设置手动提交
            connection.setAutoCommit(false);
            UserMapper mapper = sqlSession.getMapper(UserMapper.class);
            // 先做删除操作,如果子线程出现异常,此操作会回滚
            mapper.delete(null);
            // 获取线程池
            ExecutorService service = ExecutorConfig.getThreadPool();
            // 拆分数据,拆分5份
            List<List<User>> lists = averageAssign(userList, 5);
            // 执行的线程 这样可以获取返回结果 而且可以阻塞线程 代替了其他方案的 CountDownLatch + AtomicBoolean 组合的功能
            List<Callable<Integer>> callableList = new ArrayList<>();
            // 监控子线程执行完毕,再执行主线程,要不然会导致主线程关闭,子线程也会随着关闭
            // CountDownLatch countDownLatch = new CountDownLatch(lists.size());
            // AtomicBoolean atomicBoolean = new AtomicBoolean(true);

            for (int i = 0; i < lists.size(); i++) {
                List<User> list = lists.get(i);
                Callable<Integer> callable = () -> mapper.batchInsertUser(list);
                callableList.add(callable);
            }
            // 执行子线程
            List<Future<Integer>> futures = service.invokeAll(callableList);
            int size = 0;
            for (Future<Integer> future : futures) {
                // 阻塞线程 等待子线程返回结果
                if (future.get() <= 0) {
                    connection.rollback();
                }
                size += future.get();
            }
            connection.commit();
            log.info("添加完毕,共计添加 {} 条", size);
        } catch (Exception e) {
            connection.rollback();
            log.info("error", e);
            throw new RuntimeException("002");
        }
    }

    /**
     * 平均拆分list方法.
     *
     * @param source
     * @param n
     * @param <T>
     * @return
     */
    public static <T> List<List<T>> averageAssign(List<T> source, int n) {
        List<List<T>> result = new ArrayList<List<T>>();
        int remaider = source.size() % n;
        int number = source.size() / n;
        int offset = 0;//偏移量
        for (int i = 0; i < n; i++) {
            List<T> value = null;
            if (remaider > 0) {
                value = source.subList(i * number + offset, (i + 1) * number + offset + 1);
                remaider--;
                offset++;
            } else {
                value = source.subList(i * number + offset, (i + 1) * number + offset);
            }
            result.add(value);
        }
        return result;
    }
}
